package net.flyinggroup.kafka.stand;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpStatus;
import org.springframework.integration.annotation.Poller;
import org.springframework.messaging.Message;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseStatus;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;

import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

import static java.time.Instant.now;

/**
 * @author Walker Zhang
 */
@SpringBootApplication
@Controller
@Slf4j
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, "--spring.cloud.function.definition=stringSupplier");
    }

    @Bean
    public Consumer<Message<?>> avroMsg() {
        return value -> {
            System.out.println("Received: " + value);
        };
    }


    @Bean
    public Function<String, String> uppercase() {
        return value -> {
            System.out.println("Received: " + value);
            return value.toUpperCase();
        };
    }

    @Bean
    public Function<Tuple2<Flux<String>, Flux<Integer>>, Flux<String>> gather() {
        return tuple -> {
            Flux<String> stringStream = tuple.getT1();
            Flux<String> intStream = tuple.getT2().map(String::valueOf);
            return Flux.merge(stringStream, intStream);
        };
    }

    @Bean
    public Function<Flux<String>, Flux<String>> reactiveUpperCase() {
        return flux -> flux.map(val -> val.toUpperCase());
    }

    @Bean
    public Function<String, String> wrapInQuotes() {
        return s -> "\"" + s + "\"";
    }

    @Bean
    public Supplier<Flux<String>> stringSupplier() {
        return () -> {
            log.info("Trigger at: " + now());
            return Flux.just("1", "2", "3");
        };
    }

    @Autowired
    private StreamBridge bridge;

    @RequestMapping
    @ResponseStatus(HttpStatus.ACCEPTED)
    public String delegateToSupplier(@RequestBody String body) {
        System.out.println("Sending " + body);
        return body;
    }
}